package org.netwia.networks;

import java.io.IOException;
import java.io.InputStream;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.servlet.MultipartConfigElement;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.Part;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.handler.AbstractHandler;
import org.netwia.utils.DomainUtils;
import org.netwia.utils.StartStopService;
import org.netwia.utils.ValidationUtils;

/**
 * This is server responsible for package sending.
 * Packages are simply taken as they are and forwarded to the appropriate client.
 * This implementation expects fixed set of client names defined during the construction.
 * Checking whether client is alive or not is not supported in this implementation.
 * 
 * @author radek.hecl
 *
 */
public class FixedHttpPackageServer implements StartStopService {

    /**
     * Builder class.
     */
    public static class Builder {

        /**
         * Server port.
         */
        private Integer port;

        /**
         * Names of the clients.
         */
        private Set<String> clientNames = new HashSet<String>();

        /**
         * Sets server port.
         * 
         * @param port server port
         * @return this instance
         */
        public Builder setPort(int port) {
            this.port = port;
            return this;
        }

        /**
         * Sets client names.
         * 
         * @param clientNames client names
         * @return this instance
         */
        public Builder setClientNames(Collection<String> clientNames) {
            this.clientNames = DomainUtils.softCopySet(clientNames);
            return this;
        }

        /**
         * Adds new client into the client names.
         * 
         * @param clientName client name
         * @return this instance
         */
        public Builder addClientName(String clientName) {
            clientNames.add(clientName);
            return this;
        }

        /**
         * Builds the result object.
         * 
         * @return created object
         */
        public FixedHttpPackageServer build() {
            return new FixedHttpPackageServer(this);
        }
    }

    /**
     * Server port.
     */
    private Integer port;

    /**
     * Names of the clients.
     */
    private Set<String> clientNames;

    /**
     * Lock object.
     */
    private Lock lock = new ReentrantLock();

    /**
     * Flag indicating whether processing is running or not.
     */
    private boolean running = false;

    /**
     * Server instance.
     */
    private Server server;

    /**
     * Queues. Each queue is for 1 client.
     */
    private Map<String, Queue<byte[]>> queues;

    /**
     * Creates new instance.
     * 
     * @param builder builder object
     */
    public FixedHttpPackageServer(Builder builder) {
        port = builder.port;
        clientNames = Collections.unmodifiableSet(DomainUtils.softCopySet(builder.clientNames));
        guardInvariants();
        initialize();
    }

    /**
     * Guards this object to be consistent. Throws exception if this is not the case.
     */
    private void guardInvariants() {
        ValidationUtils.guardNotNull(port, "port cannot be null");
        ValidationUtils.guardNotEmptyStringInCollection(clientNames, "clientNames cannot have empty string");
        ValidationUtils.guardPositiveInt(clientNames.size(), "clientNames must have at least 1 element");
    }

    /**
     * Makes all the necessary initialization.
     */
    private void initialize() {
        queues = new HashMap<String, Queue<byte[]>>(clientNames.size());
        for (String client : clientNames) {
            queues.put(client, new LinkedBlockingQueue<byte[]>());
        }
        server = new Server(port);
        server.setHandler(new MessageHandler.Builder().
                setOwner(this).
                build());
    }

    @Override
    public void start() {
        lock.lock();
        try {
            if (running) {
                return;
            }
            server.start();
            Thread.sleep(500);
            running = true;
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void stop() {
        lock.lock();
        try {
            if (!running) {
                return;
            }
            running = false;
            server.stop();
            Thread.sleep(500);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public String toString() {
        return ToStringBuilder.reflectionToString(this);
    }

    /**
     * Server message handler.
     */
    private static class MessageHandler extends AbstractHandler {

        /**
         * Static configuration to be able use multipart requests.
         * Jetty has troubles if it is not here.
         */
        private static final MultipartConfigElement MULTI_PART_CONFIG = new MultipartConfigElement(System.getProperty("java.io.tmpdir"));

        /**
         * Builder class.
         */
        public static class Builder {

            /**
             * Owner of this object.
             */
            private FixedHttpPackageServer owner;

            /**
             * Sets owner of this object.
             * 
             * @param owner owner
             * @return this instance
             */
            public Builder setOwner(FixedHttpPackageServer owner) {
                this.owner = owner;
                return this;
            }

            /**
             * Builds the result object.
             * 
             * @return created object
             */
            public MessageHandler build() {
                return new MessageHandler(this);
            }
        }

        /**
         * Owner of this object.
         */
        private FixedHttpPackageServer owner;

        /**
         * List with random clients.
         */
        private ArrayList<String> randomClients;
        
        /**
         * Index for the next random client.
         */
        private int randomIndex = 0;
        
        /**
         * Creates new instance
         * 
         * @param builder builder object
         */
        public MessageHandler(Builder builder) {
            owner = builder.owner;
            guardInvariants();
            randomClients = new ArrayList<String>(owner.clientNames);
        }

        /**
         * Guards this object to be consistent. Throws exception if this is not the case.
         */
        private void guardInvariants() {
            ValidationUtils.guardNotNull(owner, "owner cannot be null");
        }

        @Override
        public void handle(String path, Request req, HttpServletRequest request, HttpServletResponse response)
                throws IOException, ServletException {
            Map<String, List<Object>> parameters = new HashMap<String, List<Object>>();
            if (request.getContentType() != null && request.getContentType().startsWith("multipart/form-data")) {
                request.setAttribute(Request.__MULTIPART_CONFIG_ELEMENT, MULTI_PART_CONFIG);
                for (Part part : request.getParts()) {
                    String name = part.getName();
                    if (!parameters.containsKey(name)) {
                        parameters.put(name, new ArrayList<Object>());
                    }
                    Object data = null;
                    if (part.getContentType().matches("^text/plain.*$")) {
                        // discover encoding
                        String encoding = null;
                        Pattern p = Pattern.compile("^.*charset=([^;]+);?.*$");
                        Matcher m = p.matcher(part.getContentType());
                        if (m.find()) {
                            encoding = m.group(1);
                        }
                        // convert stream
                        InputStream is = null;
                        try {
                            is = part.getInputStream();
                            StringWriter writer = new StringWriter();
                            IOUtils.copy(is, writer, encoding);
                            data = writer.toString();
                            is.close();
                        } finally {
                            IOUtils.closeQuietly(is);
                        }
                    }
                    else {
                        part.getSize();
                        InputStream is = null;
                        try {
                            is = part.getInputStream();
                            byte[] bytes = new byte[(int) part.getSize()];
                            is.read(bytes);
                            is.close();
                            data = bytes;
                        } finally {
                            IOUtils.closeQuietly(is);
                        }
                    }
                    parameters.get(name).add(data);
                }
            }
            else {
                @SuppressWarnings("unchecked")
                Enumeration<String> enm = (Enumeration<String>) req.getParameterNames();
                while (enm.hasMoreElements()) {
                    String name = enm.nextElement();
                    parameters.put(name, Arrays.asList((Object) request.getParameterValues(name)));
                }
            }
            String senderClient = (String) (parameters.get("senderClient").get(0));
            for (int i = 0; parameters.keySet().contains("package" + i + ".bin"); ++i) {
                String target = (String) (parameters.get("package" + i + ".targetClient").get(0));
                byte[] data = (byte[]) (parameters.get("package" + i + ".bin").get(0));
                if (target.equals("*")) {
                    for (String client : owner.clientNames) {
                        if (!client.equals(senderClient)) {
                            owner.queues.get(client).add(data);
                        }
                    }
                }
                else if (target.equals("?")) {
                    if (randomClients.get(randomIndex).equals(senderClient)) {
                        ++randomIndex;
                        if (randomIndex == randomClients.size()) {
                            randomIndex = 0;
                        }
                    }
                    owner.queues.get(randomClients.get(randomIndex)).add(data);
                    ++randomIndex;
                    if (randomIndex == randomClients.size()) {
                        randomIndex = 0;
                    }
                }
                else {
                    owner.queues.get(target).add(data);
                }
            }
            int length = 0;
            List<byte[]> readyPackages = new ArrayList<byte[]>();
            Queue<byte[]> queue = owner.queues.get(senderClient);
            while (!queue.isEmpty()) {
                byte[] pckg = queue.poll();
                length = length + pckg.length;
                readyPackages.add(pckg);
            }
            response.setContentLength(length);
            if (length == 0) {
                response.getOutputStream().write(new byte[] {});
            }
            else {
                for (byte[] pckg : readyPackages) {
                    response.getOutputStream().write(pckg);
                }
            }
        }

        @Override
        public String toString() {
            return ToStringBuilder.reflectionToString(this);
        }

    }

}
